#include "U2WorkQueue.h"

#include "U2LogManager.h"
#include "U2Root.h"
#include "U2Timer.h"


U2EG_NAMESPACE_USING


//---------------------------------------------------------------------
u2uint16 WorkQueue::getChannel(const U2String& channelName)
{
	U2_LOCK_MUTEX(mChannelMapMutex)

	ChannelMap::iterator i = mChannelMap.find(channelName);
	if (i == mChannelMap.end())
	{
		i = mChannelMap.insert(ChannelMap::value_type(channelName, mNextChannel++)).first;
	}
	return i->second;
}
//---------------------------------------------------------------------
WorkQueue::Request::Request(u2uint16 channel, u2uint16 rtype, const Any& rData, u2uint8 retry, RequestID rid)
	: mChannel(channel), mType(rtype), mData(rData), mRetryCount(retry), mID(rid), mAborted(false)
{

}
//---------------------------------------------------------------------
WorkQueue::Request::~Request()
{

}
//---------------------------------------------------------------------
//---------------------------------------------------------------------
WorkQueue::Response::Response(const Request* rq, bool success, const Any& data, const U2String& msg)
	: mRequest(rq), mSuccess(success), mMessages(msg), mData(data)
{
	
}
//---------------------------------------------------------------------
WorkQueue::Response::~Response()
{
	U2_DELETE mRequest;
}
//---------------------------------------------------------------------
//---------------------------------------------------------------------
U2DefaultWorkQueueBase::U2DefaultWorkQueueBase(const U2String& name)
	: mName(name)
	, mWorkerThreadCount(1)
	, mWorkerRenderSystemAccess(false)
	, mIsRunning(false)
	, mResposeTimeLimitMS(8)
	, mWorkerFunc(0)
	, mRequestCount(0)
	, mPaused(false)
	, mAcceptRequests(true)
{
}
//---------------------------------------------------------------------
const U2String& U2DefaultWorkQueueBase::getName() const
{
	return mName;
}
//---------------------------------------------------------------------
size_t U2DefaultWorkQueueBase::getWorkerThreadCount() const
{
	return mWorkerThreadCount;
}
//---------------------------------------------------------------------
void U2DefaultWorkQueueBase::setWorkerThreadCount(size_t c)
{
	mWorkerThreadCount = c;
}
//---------------------------------------------------------------------
bool U2DefaultWorkQueueBase::getWorkersCanAccessRenderSystem() const
{
	return mWorkerRenderSystemAccess;
}
//---------------------------------------------------------------------
void U2DefaultWorkQueueBase::setWorkersCanAccessRenderSystem(bool access)
{
	mWorkerRenderSystemAccess = access;
}
//---------------------------------------------------------------------
U2DefaultWorkQueueBase::~U2DefaultWorkQueueBase()
{
	//shutdown(); // can't call here; abstract function

	for (RequestQueue::iterator i = mRequestQueue.begin(); i != mRequestQueue.end(); ++i)
	{
		U2_DELETE (*i);
	}
	mRequestQueue.clear();

	for (ResponseQueue::iterator i = mResponseQueue.begin(); i != mResponseQueue.end(); ++i)
	{
		U2_DELETE (*i);
	}
	mResponseQueue.clear();
}
//---------------------------------------------------------------------
void U2DefaultWorkQueueBase::addRequestHandler(u2uint16 channel, RequestHandler* rh)
{
	U2_LOCK_RW_MUTEX_WRITE(mRequestHandlerMutex);

	RequestHandlerListByChannel::iterator i = mRequestHandlers.find(channel);
	if (i == mRequestHandlers.end())
		i = mRequestHandlers.insert(RequestHandlerListByChannel::value_type(channel, RequestHandlerList())).first;

	RequestHandlerList& handlers = i->second;
	bool duplicate = false;
	for (RequestHandlerList::iterator j = handlers.begin(); j != handlers.end(); ++j)
	{
		if ((*j)->getHandler() == rh)
		{
			duplicate = true;
			break;
		}
	}
	if (!duplicate)
		handlers.push_back(RequestHandlerHolderPtr(U2_NEW RequestHandlerHolder(rh)));

}
//---------------------------------------------------------------------
void U2DefaultWorkQueueBase::removeRequestHandler(u2uint16 channel, RequestHandler* rh)
{
	U2_LOCK_RW_MUTEX_WRITE(mRequestHandlerMutex);

	RequestHandlerListByChannel::iterator i = mRequestHandlers.find(channel);
	if (i != mRequestHandlers.end())
	{
		RequestHandlerList& handlers = i->second;
		for (RequestHandlerList::iterator j = handlers.begin(); j != handlers.end(); ++j)
		{
			if ((*j)->getHandler() == rh)
			{
				// Disconnect - this will make it safe across copies of the list
				// this is threadsafe and will wait for existing processes to finish
				(*j)->disconnectHandler();
				handlers.erase(j);	
				break;
			}
		}

	}

}
//---------------------------------------------------------------------
void U2DefaultWorkQueueBase::addResponseHandler(u2uint16 channel, ResponseHandler* rh)
{
	ResponseHandlerListByChannel::iterator i = mResponseHandlers.find(channel);
	if (i == mResponseHandlers.end())
		i = mResponseHandlers.insert(ResponseHandlerListByChannel::value_type(channel, ResponseHandlerList())).first;

	ResponseHandlerList& handlers = i->second;
	if (std::find(handlers.begin(), handlers.end(), rh) == handlers.end())
		handlers.push_back(rh);
}
//---------------------------------------------------------------------
void U2DefaultWorkQueueBase::removeResponseHandler(u2uint16 channel, ResponseHandler* rh)
{
	ResponseHandlerListByChannel::iterator i = mResponseHandlers.find(channel);
	if (i != mResponseHandlers.end())
	{
		ResponseHandlerList& handlers = i->second;
		ResponseHandlerList::iterator j = std::find(
			handlers.begin(), handlers.end(), rh);
		if (j != handlers.end())
			handlers.erase(j);

	}
}
//---------------------------------------------------------------------
WorkQueue::RequestID U2DefaultWorkQueueBase::addRequest(u2uint16 channel, u2uint16 requestType, 
	const Any& rData, u2uint8 retryCount, bool forceSynchronous)
{
	Request* req = 0;
	RequestID rid = 0;

	{
		// lock to acquire rid and push request to the queue
		U2_LOCK_MUTEX(mRequestMutex)

		if (!mAcceptRequests || mShuttingDown)
			return 0;

		rid = ++mRequestCount;
		req = U2_NEW Request(channel, requestType, rData, retryCount, rid);

		U2LogManager::getSingleton().stream(LML_TRIVIAL) << 
			"DefaultWorkQueueBase('" << mName << "') - QUEUED(thread:" <<
#if U2_THREAD_SUPPORT
			U2_THREAD_CURRENT_ID
#else
			"main"
#endif
			<< "): ID=" << rid
		    << " channel=" << channel << " requestType=" << requestType;
#if U2_THREAD_SUPPORT
		if (!forceSynchronous)
		{
			mRequestQueue.push_back(req);
			notifyWorkers();
			return rid;
		}
#endif
	}
	
	processRequestResponse(req, true);

	return rid;
}
//---------------------------------------------------------------------
void U2DefaultWorkQueueBase::addRequestWithRID(WorkQueue::RequestID rid, u2uint16 channel, 
	u2uint16 requestType, const Any& rData, u2uint8 retryCount)
{
	// lock to push request to the queue
	U2_LOCK_MUTEX(mRequestMutex)

	if (mShuttingDown)
		return;

	Request* req = U2_NEW Request(channel, requestType, rData, retryCount, rid);

	U2LogManager::getSingleton().stream(LML_TRIVIAL) << 
		"DefaultWorkQueueBase('" << mName << "') - REQUEUED(thread:" <<
#if U2_THREAD_SUPPORT
		U2_THREAD_CURRENT_ID
#else
		"main"
#endif
		<< "): ID=" << rid
			   << " channel=" << channel << " requestType=" << requestType;
#if U2_THREAD_SUPPORT
	mRequestQueue.push_back(req);
	notifyWorkers();
#else
	processRequestResponse(req, true);
#endif
}
//---------------------------------------------------------------------
void U2DefaultWorkQueueBase::abortRequest(RequestID id)
{
	U2_LOCK_MUTEX(mProcessMutex)

	// NOTE: Pending requests are exist any of RequestQueue, ProcessQueue and
	// ResponseQueue when keeping ProcessMutex, so we check all of these queues.

	for (RequestQueue::iterator i = mProcessQueue.begin(); i != mProcessQueue.end(); ++i)
	{
		if ((*i)->getID() == id)
		{
			(*i)->abortRequest();
			break;
		}
	}

	{
		U2_LOCK_MUTEX(mRequestMutex)

		for (RequestQueue::iterator i = mRequestQueue.begin(); i != mRequestQueue.end(); ++i)
		{
			if ((*i)->getID() == id)
			{
				(*i)->abortRequest();
				break;
			}
		}
	}

	{
		U2_LOCK_MUTEX(mResponseMutex)

		for (ResponseQueue::iterator i = mResponseQueue.begin(); i != mResponseQueue.end(); ++i)
		{
			if( (*i)->getRequest()->getID() == id )
			{
				(*i)->abortRequest();
				break;
			}
		}
	}
}
//---------------------------------------------------------------------
void U2DefaultWorkQueueBase::abortRequestsByChannel(u2uint16 channel)
{
	U2_LOCK_MUTEX(mProcessMutex)

	for (RequestQueue::iterator i = mProcessQueue.begin(); i != mProcessQueue.end(); ++i)
	{
		if ((*i)->getChannel() == channel)
		{
			(*i)->abortRequest();
		}
	}

	{
		U2_LOCK_MUTEX(mRequestMutex)

		for (RequestQueue::iterator i = mRequestQueue.begin(); i != mRequestQueue.end(); ++i)
		{
			if ((*i)->getChannel() == channel)
			{
				(*i)->abortRequest();
			}
		}
	}

	{
		U2_LOCK_MUTEX(mResponseMutex)

		for (ResponseQueue::iterator i = mResponseQueue.begin(); i != mResponseQueue.end(); ++i)
		{
			if( (*i)->getRequest()->getChannel() == channel )
			{
				(*i)->abortRequest();
			}
		}
	}
}
//---------------------------------------------------------------------
void U2DefaultWorkQueueBase::abortAllRequests()
{
	U2_LOCK_MUTEX(mProcessMutex)

	for (RequestQueue::iterator i = mProcessQueue.begin(); i != mProcessQueue.end(); ++i)
	{
		(*i)->abortRequest();
	}

	{
		U2_LOCK_MUTEX(mRequestMutex)

		for (RequestQueue::iterator i = mRequestQueue.begin(); i != mRequestQueue.end(); ++i)
		{
			(*i)->abortRequest();
		}
	}

	{
		U2_LOCK_MUTEX(mResponseMutex)

		for (ResponseQueue::iterator i = mResponseQueue.begin(); i != mResponseQueue.end(); ++i)
		{
			(*i)->abortRequest();
		}
	}
}
//---------------------------------------------------------------------
void U2DefaultWorkQueueBase::setPaused(bool pause)
{
	U2_LOCK_MUTEX(mRequestMutex)

	mPaused = pause;
}
//---------------------------------------------------------------------
bool U2DefaultWorkQueueBase::isPaused() const
{
	return mPaused;
}
//---------------------------------------------------------------------
void U2DefaultWorkQueueBase::setRequestsAccepted(bool accept)
{
	U2_LOCK_MUTEX(mRequestMutex)

	mAcceptRequests = accept;
}
//---------------------------------------------------------------------
bool U2DefaultWorkQueueBase::getRequestsAccepted() const
{
	return mAcceptRequests;
}
//---------------------------------------------------------------------
void U2DefaultWorkQueueBase::_processNextRequest()
{
	Request* request = 0;
	{
		// scoped to only lock while retrieving the next request
		U2_LOCK_MUTEX(mProcessMutex)
		{
			U2_LOCK_MUTEX(mRequestMutex)

			if (!mRequestQueue.empty())
			{
				request = mRequestQueue.front();
				mRequestQueue.pop_front();
				mProcessQueue.push_back( request );
			}
		}
	}

	if (request)
	{
		processRequestResponse(request, false);
	}


}
//---------------------------------------------------------------------
void U2DefaultWorkQueueBase::processRequestResponse(Request* r, bool synchronous)
{
	Response* response = processRequest(r);

	U2_LOCK_MUTEX(mProcessMutex)

	RequestQueue::iterator it;
	for( it = mProcessQueue.begin(); it != mProcessQueue.end(); ++it )
	{
		if( (*it) == r )
		{
			mProcessQueue.erase( it );
			break;
		}
	}

	if (response)
	{
		if (!response->succeeded())
		{
			// Failed, should we retry?
			const Request* req = response->getRequest();
			if (req->getRetryCount())
			{
				addRequestWithRID(req->getID(), req->getChannel(), req->getType(), req->getData(), 
					req->getRetryCount() - 1);
				// discard response (this also deletes request)
				U2_DELETE response;
				return;
			}
		}
		if (synchronous)
		{
			processResponse(response);
			U2_DELETE response;
		}
		else
		{
			if( response->getRequest()->getAborted() )
			{
				// destroy response user data
				response->abortRequest();
			}
			// Queue response
			U2_LOCK_MUTEX(mResponseMutex)
			mResponseQueue.push_back(response);
			// no need to wake thread, this is processed by the main thread
		}

	}
	else
	{
		// no response, delete request
		U2LogManager::getSingleton().stream() << 
			"DefaultWorkQueueBase('" << mName << "') warning: no handler processed request "
			<< r->getID() << ", channel " << r->getChannel()
			<< ", type " << r->getType();
		U2_DELETE r;
	}

}
//---------------------------------------------------------------------
void U2DefaultWorkQueueBase::processResponses() 
{
	unsigned long msStart = U2Root::getSingleton().getTimer()->getMilliseconds();
	unsigned long msCurrent = 0;

	// keep going until we run out of responses or out of time
	while(true)
	{
		Response* response = 0;
		{
			U2_LOCK_MUTEX(mResponseMutex)

			if (mResponseQueue.empty())
				break; // exit loop
			else
			{
				response = mResponseQueue.front();
				mResponseQueue.pop_front();
			}
		}

		if (response)
		{
			processResponse(response);
			U2_DELETE response;
		}

		// time limit
		if (mResposeTimeLimitMS)
		{
			msCurrent = U2Root::getSingleton().getTimer()->getMilliseconds();
			if (msCurrent - msStart > mResposeTimeLimitMS)
				break;
		}
	}
}
//---------------------------------------------------------------------
WorkQueue::Response* U2DefaultWorkQueueBase::processRequest(Request* r)
{
	RequestHandlerListByChannel handlerListCopy;
	{
		// lock the list only to make a copy of it, to maximise parallelism
		U2_LOCK_RW_MUTEX_READ(mRequestHandlerMutex);
		
		handlerListCopy = mRequestHandlers;
		
	}

	Response* response = 0;

	U2StringUtil::U2StrStreamType dbgMsg;
	dbgMsg <<
#if U2_THREAD_SUPPORT
		U2_THREAD_CURRENT_ID
#else
		"main"
#endif
		<< "): ID=" << r->getID() << " channel=" << r->getChannel() 
		<< " requestType=" << r->getType();

	U2LogManager::getSingleton().stream(LML_TRIVIAL) << 
		"DefaultWorkQueueBase('" << mName << "') - PROCESS_REQUEST_START(" << dbgMsg.str();

	RequestHandlerListByChannel::iterator i = handlerListCopy.find(r->getChannel());
	if (i != handlerListCopy.end())
	{
		RequestHandlerList& handlers = i->second;
		for (RequestHandlerList::reverse_iterator j = handlers.rbegin(); j != handlers.rend(); ++j)
		{
			// threadsafe call which tests canHandleRequest and calls it if so 
			response = (*j)->handleRequest(r, this);

			if (response)
				break;
		}
	}

	U2LogManager::getSingleton().stream(LML_TRIVIAL) << 
		"DefaultWorkQueueBase('" << mName << "') - PROCESS_REQUEST_END(" << dbgMsg.str()
		<< " processed=" << (response!=0);

	return response;

}
//---------------------------------------------------------------------
void U2DefaultWorkQueueBase::processResponse(Response* r)
{
	U2StringUtil::U2StrStreamType dbgMsg;
	dbgMsg << "thread:" <<
#if U2_THREAD_SUPPORT
		U2_THREAD_CURRENT_ID
#else
		"main"
#endif
		<< "): ID=" << r->getRequest()->getID()
		<< " success=" << r->succeeded() << " messages=[" << r->getMessages() << "] channel=" 
		<< r->getRequest()->getChannel() << " requestType=" << r->getRequest()->getType();

	U2LogManager::getSingleton().stream(LML_TRIVIAL) << 
		"DefaultWorkQueueBase('" << mName << "') - PROCESS_RESPONSE_START(" << dbgMsg.str();

	ResponseHandlerListByChannel::iterator i = mResponseHandlers.find(r->getRequest()->getChannel());
	if (i != mResponseHandlers.end())
	{
		ResponseHandlerList& handlers = i->second;
		for (ResponseHandlerList::reverse_iterator j = handlers.rbegin(); j != handlers.rend(); ++j)
		{
			if ((*j)->canHandleResponse(r, this))
			{
				(*j)->handleResponse(r, this);
			}
		}
	}
	U2LogManager::getSingleton().stream(LML_TRIVIAL) << 
		"DefaultWorkQueueBase('" << mName << "') - PROCESS_RESPONSE_END(" << dbgMsg.str();

}
//---------------------------------------------------------------------

void U2DefaultWorkQueueBase::WorkerFunc::operator()()
{
	mQueue->_threadMain();
}

void U2DefaultWorkQueueBase::WorkerFunc::run()
{
	mQueue->_threadMain();
}
